package com.service.websocket.service;

import cn.hutool.extra.spring.SpringUtil;
import com.service.websocket.handler.AbstractMessageHandler;
import com.service.websocket.exception.WebSocketParamsException;
import com.service.websocket.to.MessageTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @author xianggj
 * @description: 任务处理器 用于接收任务和分发任务
 * @date 2019/8/21 17:10
 */
@Component
@Slf4j
public class WebSocketTaskHandler {

    /**
     * 初始化缓存对象
     */
    public static final Map<String, AbstractMessageHandler> PRECESSOR_MAP = new ConcurrentHashMap<>();

    /**
     * 执行缓存
     * 格式：1.usrId客户端ID 2.params 参数列表 3.UUID请求唯一标示 4.处理器 5.推送对象
     */
    public static final ConcurrentLinkedQueue<MessageTO> TASK_QUEUE = new ConcurrentLinkedQueue<>();

    /**
     * 执行频率 (单位毫秒)
     */
    public static final int EXE_FREQUENCY  = 1000;


    /**
     * 这里最好延迟加载
     * 初始化加载有个资源浪费问题 当关闭连接时回去扫描没有调用的类
     */
    public void afterPropertiesSet(){
        Map<String, AbstractMessageHandler> beansOfType = SpringUtil.getBeansOfType(AbstractMessageHandler.class);
        for (Map.Entry<String, AbstractMessageHandler> entry: beansOfType.entrySet()) {
            log.info("TaskHandler开始装载:" + entry.getKey());
            AbstractMessageHandler processor = entry.getValue();
            if (processor == null){
                continue;
            }
            PRECESSOR_MAP.put(processor.getTopic(), processor);
        }
    }

    /**
     * 推送数据给前端
     * 推送时判断状态：
     *  1.如果状态为已完成或者失败，则只推一次
     *  2.如果状态为运行中，则循环推送
     * 开始推送
     * @param poll
     */
    private void startPush(MessageTO poll) {
        if (poll == null){
            return;
        }
        AbstractMessageHandler messageProcessor = poll.getAmp();
        try {
            //生产消息
            messageProcessor.produce(poll);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("产生数据失败！" + e.getMessage());
        }
    }



    /**
     * 一周期休息一次
     * @param cycleFlag
     * @param winId
     */
    private void cycle(Set<String> cycleFlag, String winId) {
        if (cycleFlag.contains(winId)){
            //轮询完一轮了 休息一秒钟 等消息发送者发送消息
            try {
                Thread.sleep(EXE_FREQUENCY);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            cycleFlag.clear();
        } else {
            cycleFlag.add(winId);
        }
    }

    /**
     * 多个线程的特殊处理
     */
    private ExecutorService scheduler = null;

    /**
     * 维持一个线程池总是工作
     */
    private void checkOrCreateScheduler(MessageTO tuple) {
        if (scheduler == null || scheduler.isShutdown()) {
            scheduler =  new ThreadPoolExecutor(20, 20,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
        }
        /**
         * 添加线程执行
         */
        scheduler.execute(()-> {
            try {
                startPush(tuple);
            } catch (Exception e) {
                log.error("获取数据失败！" + e.getMessage());
                e.printStackTrace();
            }
        });
    }


    /**
     * 添加到执行队列 开始执行
     * @param messageTO 消息体
     */
    public void addPushQueue(MessageTO messageTO) {
        if (PRECESSOR_MAP.isEmpty()){
            try {
                afterPropertiesSet();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        AbstractMessageHandler amp = PRECESSOR_MAP.get(messageTO.getTopic());
        //这个错误概率比较低
        if (amp == null){
            String errorMsg = "错误：没发现topic: " + messageTO.getTopic();
            log.error(errorMsg);
            return ;
        }
        try {
            Object data =  messageTO.getData();
            //参数预处理
            Object params = amp.getParams(data);
            messageTO.setData(params);
        } catch (WebSocketParamsException we) {
            try {
                amp.sendErrorMessage(messageTO, "参数错误："+we.getMessage());
            } catch (Exception e1) {
                e1.printStackTrace();
            }
            return;
        } catch (Exception e) {
            try {
                amp.sendErrorMessage(messageTO, "参数检验时出现了错误："+e.getMessage());
            } catch (Exception e1) {
                e1.printStackTrace();
            }
            return;
        }
        messageTO.setAmp(amp);
        checkOrCreateScheduler(messageTO);
    }


    /**
     * 清除队列中过时的key
     * 关闭某窗口时调用此方法 清除正在执行中的任务
     * @param winId
     * @return
     */
    public void removeKeyByWinId(String winId){

        Iterator<MessageTO> iterator = TASK_QUEUE.iterator();
        while (iterator.hasNext()) {
            MessageTO peek = iterator.next();
            String s = peek.getWindowId();
            if (s.equals(winId)){
                TASK_QUEUE.remove(peek);
            }
        }
    }
}
